package com.piece.core.kafka.producer;

import com.piece.core.framework.constant.RegisterConstants;
import com.piece.core.framework.support.dto.PayloadDTO;
import com.piece.core.framework.util.message.Message;
import com.piece.core.framework.util.message.MessageFactory;
import com.piece.core.framework.util.string.JsonUtil;
import com.piece.core.framework.util.string.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Slf4j
public class KafkaMessageProducer implements Message {

    private MessageFactory messageFactory;
    private KafkaTemplate kafkaTemplate;

    public KafkaMessageProducer(MessageFactory messageFactory, KafkaTemplate kafkaTemplate) {
        this.messageFactory = messageFactory;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Override
    public boolean isDefault() {
        return false;
    }

    @Override
    public void send(PayloadDTO payload) throws Exception {
        messageFactory.getOperator(payload.getOperator()).prepareSend(payload, "kafkaMessageProducer");
        convertAndSend(payload);
    }

    @Override
    public void reSend(PayloadDTO payload) throws Exception {
        if (messageFactory.getOperator(payload.getOperator()).checkRetryLimit(payload)) {
            convertAndSend(payload);
        }
    }

    private void convertAndSend(PayloadDTO payload) {
        String topic = StringUtil.isEmpty(payload.getTopic()) ? RegisterConstants.TOPIC : payload.getTopic();
        kafkaTemplate.send(topic, JsonUtil.toJson(payload)).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.debug("消息发送失败:topic({}),payload({}),cause({})", topic, JsonUtil.toJson(payload), ex);
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.debug("消息发送成功:topic({}),result({})", topic, result);
            }
        });
    }
}
